package com.example.java.cep;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;

/**
 * 24小时内，至少5次有效交易的用户
 */
public class ActiveUserDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        DataStreamSource<ActiveUserBean> data = env.fromElements(
                new ActiveUserBean("100XX", 0.0D, 1597905234000L),
                new ActiveUserBean("100XX", 100.0D, 1597905235000L),
                new ActiveUserBean("100XX", 200.0D, 1597905236000L),
                new ActiveUserBean("100XX", 300.0D, 1597905237000L),
                new ActiveUserBean("100XX", 400.0D, 1597905238000L),
                new ActiveUserBean("100XX", 500.0D, 1597905239000L),
                new ActiveUserBean("101XX", 0.0D, 1597905240000L),
                new ActiveUserBean("101XX", 100.0D, 1597905241000L)
        );
        SingleOutputStreamOperator<ActiveUserBean> watermark = data
                .assignTimestampsAndWatermarks(new WatermarkStrategy<ActiveUserBean>() {
                    @Override
                    public WatermarkGenerator<ActiveUserBean> createWatermarkGenerator(
                            WatermarkGeneratorSupplier.Context context) {
                        return new WatermarkGenerator<ActiveUserBean>() {
                            private long maxTimeStamp = Long.MIN_VALUE;
                            private long maxOutOfOrderness = 500L;

                            @Override
                            public void onEvent(ActiveUserBean event, long eventTimestamp, WatermarkOutput output) {
                                maxTimeStamp = Math.max(maxTimeStamp, event.getTimestamp());
                            }


                            @Override
                            public void onPeriodicEmit(WatermarkOutput output) {
                                output.emitWatermark(new Watermark(maxTimeStamp - maxOutOfOrderness));
                            }
                        };
                    }
                }.withTimestampAssigner(((element, recordTimestamp) -> element.getTimestamp())));
        KeyedStream<ActiveUserBean, String> keyed = watermark.keyBy(value -> value.getUid());
        Pattern<ActiveUserBean, ActiveUserBean> pattern = Pattern
                .<ActiveUserBean>begin("start")
                .where(new SimpleCondition<ActiveUserBean>() {
                    @Override
                    public boolean filter(ActiveUserBean value) throws Exception {
                        return value.getPayMoney() > 0;
                    }
                })
                .timesOrMore(5)
                .within(Time.hours(24));

        PatternStream<ActiveUserBean> patternStream = CEP.pattern(keyed, pattern);
        SingleOutputStreamOperator<ActiveUserBean> process = patternStream.process(
                new PatternProcessFunction<ActiveUserBean, ActiveUserBean>() {
                    @Override
                    public void processMatch(Map<String, List<ActiveUserBean>> match, Context ctx,
                                             Collector<ActiveUserBean> out) throws Exception {
                        System.out.println(match);
                    }
                });
        process.print();
        env.execute();

    }
}
